using System;
using System.Collections.Generic;
using System.Threading;
using Me.Prettyprint.Cassandra.Connection;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Ddl;
using Me.Prettyprint.Hector.Api.Exceptions;
using Org.Apache.Cassandra.Thrift;
using Thrift;

namespace Me.Prettyprint.Cassandra.Service
{
	/// <summary>
	/// Base class for Cassandra cluster. A couple of important notes. When Cassandra
    /// starts up, it can take up to 30 seconds for the cluster to stabilize. The
    /// class sets the default ring delay to 30 seconds. When schema changes are made
    /// it may take a few seconds for schema agreement, so there's a delay of 1 second
    /// to give it a chance to sync up.
	/// </summary>
	public abstract class AbstractCluster : Cluster
	{
        protected Logger logger = new Logger(typeof(AbstractCluster));
        public static Dictionary<string,string> EMPTY_CREDENTIALS = new Dictionary<string,string>();
        public const int RING_DELAY = 30 * 1000; // delay after which we assume ring has stabilized
        public const int WAIT_FOR_SCHEMA_AGREEMENT_SLEEP_TIME = 1000;
        protected CassandraClientMonitor cassandraClientMonitor;
        protected List<CassandraHost> knownPoolHosts;
        protected readonly ExceptionsTranslator exceptionTranslator;

        delegate void DescribeClusterNameDel();
        delegate void DescribeKeyspaceDel(string ksname);
        delegate void DescribeKeyspacesDel();
        delegate void DescribePartitionerDel();
        delegate void DescribeRingDel(string ksname);
        delegate void DescribeSchemaVersionsDel();
        delegate void DescribeThriftVersionDel();

		public AbstractCluster(string clusterName, CassandraHostConfigurator cassandraHostConfigurator) : this(clusterName, cassandraHostConfigurator, EMPTY_CREDENTIALS) 
		{
		}
        public AbstractCluster(string clusterName, CassandraHostConfigurator cassandraHostConfigurator, Dictionary<string,string> creds)
        {
            ConnectionManager = new HConnectionManager(clusterName, cassandraHostConfigurator);
            Name = clusterName;
            Configurator = cassandraHostConfigurator;
            FailoverPolicy = FailoverPolicy.ON_FAIL_TRY_ALL_AVAILABLE;
            exceptionTranslator = new ExceptionsTranslatorImpl();
            ClockResolution = CassandraHostConfigurator.GetClockResolution();
            this.cassandraClientMonitor = new CassandraClientMonitor(ConnectionManager);
            ConnectionManager.SetCassandraClientMonitor(this.cassandraClientMonitor);
            if (Credentials == null || Credentials.Count == 0)
            {
                Credentials = EMPTY_CREDENTIALS;
            }
            else
            {
                Credentials = creds;
            }
        }
        public Dictionary<string, string> Credentials
        {
            get;
            set;
        }
        public HConnectionManager ConnectionManager
		{
			get;
            protected set;
		}
		public string Name
		{
			get;
            protected set;
		}
		public CassandraHostConfigurator Configurator
        {
            get;
            set;
        }
        public ClockResolution ClockResolution
        {
            get;
            set;
        }
        public FailoverPolicy FailoverPolicy
        {
            get;
            private set;
        }
        public CassandraClientMonitor ClientMonitor
        {
            get
            {
                return this.cassandraClientMonitor;
            }
        }
        public List<CassandraHost> KnownPoolHosts(bool refresh)
        {
            if (refresh || knownPoolHosts == null)
            {
                knownPoolHosts = ConnectionManager.GetHosts();
            }
            return knownPoolHosts;
        }
        public void AddHost(CassandraHost cassandraHost, bool skipApplyConfig)
		{
			if (!skipApplyConfig && Configurator != null)
			{
				Configurator.ApplyConfig(cassandraHost);
			}
			ConnectionManager.AddCassandraHost(cassandraHost);
		}
        /// <summary>
        /// Default implementation does nothing.
        /// </summary>
        public virtual void OnStartup()
        {
        }
        public List<KeyspaceDefinition> DescribeKeyspaces()
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, List<KeyspaceDefinition>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<KsDef> result = cassandra.describe_keyspaces();
                    return ThriftKsDef.FromThriftList(result);
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<List<KeyspaceDefinition>> op = new OperationBuilder<List<KeyspaceDefinition>>(x);
            op.OperationType = OperationType.META_READ;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public KeyspaceDefinition DescribeKeyspace(string keyspace)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, KeyspaceDefinition> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    KsDef ks = cassandra.describe_keyspace(keyspace);
                    return new ThriftKsDef(ks);
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<KeyspaceDefinition> op = new OperationBuilder<KeyspaceDefinition>(x);
            op.OperationType = OperationType.META_READ;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public string DescribeClusterName()
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    string result = cassandra.describe_cluster_name();
                    return result;
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_READ;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public string DescribePartitioner()
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    string result = cassandra.describe_partitioner();
                    return result;
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_READ;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public string DescribeThriftVersion() 
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    string result = cassandra.describe_version();
                    return result;
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_READ;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public List<TokenRange> DescribeRing(string keyspace)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, List<TokenRange>> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    List<TokenRange> result = cassandra.describe_ring(keyspace);
                    return result;
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<List<TokenRange>> op = new OperationBuilder<List<TokenRange>>(x);
            op.OperationType = OperationType.META_READ;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public string DropColumnFamily(string keyspaceName, string columnFamily)
        {
            return DropColumnFamily(keyspaceName, columnFamily, false);
        }
        public string DropColumnFamily(string keyspaceName, string columnFamily, bool waitForSchemaAgreement)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    string result = cassandra.system_drop_column_family(columnFamily);
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_WRITE;
            op.FailoverPolicy = FailoverPolicy.FAIL_FAST;
            op.KeyspaceName = keyspaceName;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public string DropKeyspace(string keyspace)
        {
            return DropKeyspace(keyspace, false);
        }
        public string DropKeyspace(string keyspace, bool waitForSchemaAgreement)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    string result = cassandra.system_drop_keyspace(keyspace);
                    if (waitForSchemaAgreement)
                    {
                        WaitForSchemaAgreement(cassandra);
                    }
                    return result;
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_WRITE;
            op.FailoverPolicy = FailoverPolicy.FAIL_FAST;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
            return op.Result;
        }
        public void Truncate(string keyspaceName, string columnFamily)
        {
            Func<Org.Apache.Cassandra.Thrift.Cassandra.Client, string> x = (Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) =>
            {
                try
                {
                    cassandra.truncate(columnFamily);
                    return null;
                }
                catch (Exception e)
                {
                    throw exceptionTranslator.translate(e);
                }
            };
            Operation<string> op = new OperationBuilder<string>(x);
            op.OperationType = OperationType.META_WRITE;
            op.FailoverPolicy = FailoverPolicy.FAIL_FAST;
            op.KeyspaceName = keyspaceName;
            op.Credentials = Credentials;
            ConnectionManager.operateWithFailover(op);
        }
        protected static void WaitForSchemaAgreement(Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra) 
        {
            int waited = 0;
            int versions = 0;
            while (versions != 1)
            {
                versions = 0;
                Dictionary<string, List<string>> schema = cassandra.describe_schema_versions();
                foreach (string key in schema.Keys)
                {
                    if (!key.Equals("UNREACHABLE"))
                    {
                        versions++;
                    }
                }
                if (versions != 1)
                {
                    Thread.Sleep(WAIT_FOR_SCHEMA_AGREEMENT_SLEEP_TIME);
                    waited += WAIT_FOR_SCHEMA_AGREEMENT_SLEEP_TIME;
                    if (waited > RING_DELAY)
                    {
                        throw new SystemException("Could not reach schema agreement in " + RING_DELAY + "ms");
                    }
                }
            }
        }
        #region Abstract Methods
        public abstract string AddKeyspace(KeyspaceDefinition ksdef);
        public abstract string AddKeyspace(KeyspaceDefinition ksdef, bool waitForSchemaAgreement);
        public abstract Dictionary<string, List<string>> DescribeSchemaVersions();
        public abstract string AddColumnFamily(ColumnFamilyDefinition cfdef);
        public abstract string AddColumnFamily(ColumnFamilyDefinition cfdef, bool waitForSchemaAgreement);
        public abstract string UpdateColumnFamily(ColumnFamilyDefinition cfdef);
        public abstract string UpdateColumnFamily(ColumnFamilyDefinition cfdef, bool waitForSchemaAgreement);
        public abstract string UpdateKeyspace(KeyspaceDefinition ksdef);
        public abstract string UpdateKeyspace(KeyspaceDefinition ksdef, bool waitForSchemaAgreement);
        #endregion
    }
}